package com.enlorenz.core.channel.activemq;

import java.util.ArrayList;
import java.util.List;

import com.enlorenz.core.base.IReqResMessageHandle;
import com.enlorenz.core.channel.IChannelEngine;
import com.enlorenz.core.engine.AbstractEngine;
import com.enlorenz.core.engine.EngineUtils;
import com.enlorenz.core.queue.proc.AbstractObjectCycleThread;
import com.enlorenz.core.queue.transfer.imp.WaitMessageQueue;

/**
 * 双通道的MQ
 * 带MQ发送通道和MQ接收通道
 * 从接收通道接收数据然后转发到发送通道
 * @author lcb
 */
public class MQRecvSendChannel extends AbstractEngine implements IChannelEngine{
	//客户端MQ通道
	private MQClientChannel clientChannel=new MQClientChannel();
	//服务端MQ通道
	private MQServerChannel serverChannel=new MQServerChannel();
	//客户端MQ URL
	private String clientMqUrl;
	//客户端MQ对列名
	private String clientMqName;
	
	//服务端MQ URL
	private String serverMqUrl;
	//服务端MQ对列名
	private String serverMqName;
	//发送队列
	private WaitMessageQueue sendQueue=new WaitMessageQueue();
	//接收队列
	private WaitMessageQueue recvQueue=new WaitMessageQueue();
	//接收消息处理器
	private IReqResMessageHandle recvMsgHandle;
	//处理接收消息的线程列表
	private List<HandleThread> handleRecvThreadList=new ArrayList<HandleThread>();
	//处理接收线程的数量
	private int handRecvThreadNum=5;
	//是否初始化过
	private boolean isInit=false;
	
	public void setHandRecvThreadNum(int handRecvThreadNum) {
		this.handRecvThreadNum = handRecvThreadNum;
	}

	public void setRecvMsgHandle(IReqResMessageHandle recvMsgHandle) {
		this.recvMsgHandle = recvMsgHandle;
	}
	public void setClientMqUrl(String clientMqUrl) {
		this.clientMqUrl = clientMqUrl;
	}

	public void setClientMqName(String clientMqName) {
		this.clientMqName = clientMqName;
	}

	public void setServerMqUrl(String serverMqUrl) {
		this.serverMqUrl = serverMqUrl;
	}

	public void setServerMqName(String serverMqName) {
		this.serverMqName = serverMqName;
	}

	/**
	 * 初始化接收线程队列
	 */
	private void initRecvThreadList(){
		if(!isInit){
			for(int i=0;i<handRecvThreadNum;i++){
				HandleThread handleThread=new HandleThread();
				handleRecvThreadList.add(handleThread);
			}
			isInit=true;
		}
	}

	@Override
	protected void initEngine() throws Exception {
		clientChannel.setName(this.getName()+"2"+clientMqName);
		clientChannel.setMqName(clientMqName);
		clientChannel.setMqUrl(clientMqUrl);
		clientChannel.setWaitMessageQueue(sendQueue);
		clientChannel.init();
		
		serverChannel.setName(clientMqName+"2"+this.getName());
		serverChannel.setMqName(serverMqName);
		serverChannel.setMqUrl(serverMqUrl);
		serverChannel.setRecvQueue(recvQueue);
		serverChannel.init();
		
		logger.info("["+this.getName()+"]初始化处理线程数量["+handRecvThreadNum+"]");
		initRecvThreadList();
		
		for(HandleThread handleRecvThread:handleRecvThreadList){
			//初始化接收线程
			handleRecvThread.init();
			handleRecvThread.setWaitMessageQueue(recvQueue);
		}
	}
	
	@Override
	public boolean isAlive() {
		if(!clientChannel.isAlive()){
			return clientChannel.isAlive();
		}
		
		if(!serverChannel.isAlive()){
			return serverChannel.isAlive();
		}
		
		return true;
	}

	

	@Override
	protected void startEngine() throws Exception {
		serverChannel.start();
		clientChannel.start();
		//启动接收线程列表
		EngineUtils.startEngineList(handleRecvThreadList);
	}

	@Override
	protected void shutdownEngine() throws Exception {
		serverChannel.shutdown();
		clientChannel.shutdown();
		//关闭接收线程列表
		EngineUtils.stopEngineList(handleRecvThreadList);
	}

	 /**
    * 处理接收消息
    * @author lcb
    */
   private class HandleThread extends AbstractObjectCycleThread{
		@Override
		protected void process(Object result) {
			Object resultMsg;
			try {
				resultMsg = recvMsgHandle.reqResMsgHandle(result);
				if(null!=resultMsg){
					sendQueue.addMsg(resultMsg);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
   }
	
	
}
